#ifndef __M_BROKER_H__
#define __M_BROKER_H__
#include "../MQthird/include/muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"
#include "../MQcommon/threadpool.hpp"
#include "../MQcommon/mq_msg.pb.h"
#include "../MQcommon/mq_proto.pb.h"
#include "../MQcommon/logger.hpp"
#include "mq_connection.hpp"
#include "mq_consumer.hpp"
#include "mq_host.hpp"
namespace bitmq{
    #define DBFILE "/meta.db"
    #define HOSTNAME "MyVirtualHost"
    class MyServer  {
    public:
    typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
    MyServer(int port, const std::string &basedir): _server(&_loop, muduo::net::InetAddress("0.0.0.0", port), "Server", muduo::net::TcpServer::kReusePort),
    _dispatcher(std::bind(&MyServer::onUnknownMessageCallBack, this, std::placeholders::_1, 
        std::placeholders::_2, std::placeholders::_3)),
    _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, 
        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
   _host(std::make_shared<VirtualHost>(HOSTNAME, basedir, basedir + DBFILE)),
    _consumers(std::make_shared<ConsumerManager>()),
    _connection_manager(std::make_shared<ConnectionManager>()),
    _pool(std::make_shared<Threadpool>())
    {
        // 针对历史消息中的所有队列，别忘了，初始化队列的消费者管理结构
        QueueMap qm = _host->allQueues();
        for (auto &q : qm)
        {
            _consumers->initQueueConsumer(q.first);
        }
        // 往分发器中注册各种回调函数
        _dispatcher.registerMessageCallback<bitmq::openChannelRequest>(std::bind(&MyServer::onOpenChannel, this,
                                                                                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<bitmq::closeChannelRequest>(std::bind(&MyServer::onCloseChannel, this,
                                                                                  std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<bitmq::declareExchangeRequest>(std::bind(&MyServer::onDeclareExchange, this,
                                                                                     std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<bitmq::deleteExchangeRequest>(std::bind(&MyServer::onDeleteExchange, this,
                                                                                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<bitmq::declareQueueRequest>(std::bind(&MyServer::onDeclareQueue, this,
                                                                                  std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<bitmq::deleteQueueRequest>(std::bind(&MyServer::onDeleteQueue, this,
                                                                                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<bitmq::queueBindRequest>(std::bind(&MyServer::onQueueBind, this,
                                                                               std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<bitmq::queueUnBindRequest>(std::bind(&MyServer::onQueueUnBind, this,
                                                                                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<bitmq::basicPublishRequest>(std::bind(&MyServer::onBasicPublish, this,
                                                                                  std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<bitmq::basicAckRequest>(std::bind(&MyServer::onBasicAck, this,
                                                                              std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<bitmq::basicConsumeRequest>(std::bind(&MyServer::onBasicConsume, this,
                                                                                  std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.registerMessageCallback<bitmq::basicCancelRequest>(std::bind(&MyServer::onBasicCancel, this,
                                                                                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        // 往IO回调函数中传递解释器的回调函数
        _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(),
                                             std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        // 传递连接建立的回调函数
        _server.setConnectionCallback(std::bind(&MyServer::onConnectionCallBack, this, std::placeholders::_1));
    }

        void start()
        {
            _server.start();
            _loop.loop();
        }

    private:
        void onUnknownMessageCallBack(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }
        //打开一个信道
        void onOpenChannel(const muduo::net::TcpConnectionPtr& conn, const openChannelRequestPtr& message, muduo::Timestamp){
             Connection::ptr tmp_ptr = _connection_manager->getConnection(conn);
             if(tmp_ptr.get() == nullptr){
                DLOG("连接不存在！");
                return ;
             }
            return tmp_ptr->openChannel(message);
        }
        //关闭信道
        void onCloseChannel(const muduo::net::TcpConnectionPtr& conn, const closeChannelRequestPtr& message, muduo::Timestamp){
            Connection::ptr tmp_ptr = _connection_manager->getConnection(conn);
             if(tmp_ptr.get() == nullptr){
                DLOG("连接不存在！");
                return ;
             }
            return tmp_ptr->closeChannel(message);
        }
        //声明交换机
        void onDeclareExchange(const muduo::net::TcpConnectionPtr& conn, const declareExchangeRequestPtr& message, muduo::Timestamp){
            Connection::ptr tmp_ptr = _connection_manager->getConnection(conn);
            if(tmp_ptr.get() == nullptr){
               DLOG("创建交换机时，没有找到指定的连接！");
               return ;
            }
           Channel::ptr _channel = tmp_ptr->getChannel(message->cid());
           if(tmp_ptr.get() == nullptr){
              DLOG("创建交换机时，没有找到指定的信道");
              return ;
           }
           return _channel->declareExchange(message);
        }
        //删除交换机
        void onDeleteExchange(const muduo::net::TcpConnectionPtr& conn, const deleteExchangeRequestPtr& message, muduo::Timestamp){
            Connection::ptr tmp_ptr = _connection_manager->getConnection(conn);
            if(tmp_ptr.get() == nullptr){
               DLOG("删除交换机时，没有找到连接对应的连接");
               return ;
            }
           Channel::ptr _channel = tmp_ptr->getChannel(message->cid());
           if(tmp_ptr.get() == nullptr){
              DLOG("删除交换机时,没有找到指定的信道");
              return ;
           }
           return _channel->deleteExchange(message);
        }
        //声明队列
        void onDeclareQueue(const muduo::net::TcpConnectionPtr& conn, const declareQueueRequestPtr& message, muduo::Timestamp){
            Connection::ptr tmp_ptr = _connection_manager->getConnection(conn);
            if(tmp_ptr.get() == nullptr){
               DLOG("创建队列时，没有找到连接对应的连接！");
               return ;
            }
           Channel::ptr _channel = tmp_ptr->getChannel(message->cid());
           if(tmp_ptr.get() == nullptr){
              DLOG("创建队列时，没有找到信道！");
              return ;
           }
           return _channel->declareQueue(message);
        }
        void onDeleteQueue(const muduo::net::TcpConnectionPtr& conn, const deleteQueueRequestPtr& message, muduo::Timestamp) {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr) {
                DLOG("删除队列时，没有找到连接对应的连接！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr) {
                DLOG("删除队列时，没有找到信道！");
                return;
            }
            return cp->deleteQueue(message);
        }
        //队列绑定
        void onQueueBind(const muduo::net::TcpConnectionPtr& conn, const queueBindRequestPtr& message, muduo::Timestamp) {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr) {
                DLOG("队列绑定时，没有找到连接对应的连接！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr) {
                DLOG("队列绑定时，没有找到信道！");
                return;
            }
            return cp->queueBind(message);
        }
        //队列解绑
        void onQueueUnBind(const muduo::net::TcpConnectionPtr& conn, const queueUnBindRequestPtr& message, muduo::Timestamp) {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr) {
                DLOG("队列解除绑定时，没有找到连接对应的连接！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr) {
                DLOG("队列解除绑定时，没有找到信道！");
                return;
            }
            return cp->queueUnBind(message);
        }
        //消息发布
        void onBasicPublish(const muduo::net::TcpConnectionPtr& conn, const basicPublishRequestPtr& message, muduo::Timestamp) {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr) {
                DLOG("发布消息时，没有找到连接对应的连接！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr) {
                DLOG("发布消息时，没有找到信道！");
                return;
            }
            return cp->basicPublish(message);
        }
        //消息确认
        void onBasicAck(const muduo::net::TcpConnectionPtr& conn, const basicAckRequestPtr& message, muduo::Timestamp) {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr) {
                DLOG("确认消息时，没有找到连接对应的连接！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr) {
                DLOG("确认消息时，没有找到信道！");
                return;
            }
            return cp->basicAck(message);
        }
        //队列消息订阅
        void onBasicConsume(const muduo::net::TcpConnectionPtr& conn, const basicConsumeRequestPtr& message, muduo::Timestamp) {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr) {
                DLOG("队列消息订阅时，没有找到连接对应的连接！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr) {
                DLOG("队列消息订阅时，没有找到信道！");
                return;
            }
            return cp->basicConsume(message);
        }
        //队列消息取消订阅
        void onBasicCancel(const muduo::net::TcpConnectionPtr& conn, const basicCancelRequestPtr& message, muduo::Timestamp) {
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr) {
                DLOG("队列消息取消订阅时，没有找到连接对应的连接！");
                conn->shutdown();
                return;
            }
            Channel::ptr cp = mconn->getChannel(message->cid());
            if (cp.get() == nullptr) {
                DLOG("队列消息取消订阅时，没有找到信道！");
                return;
            }
            return cp->basicCancel(message);
        }
        void onConnectionCallBack(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                _connection_manager->newConnection(_host,_consumers,_codec,conn,_pool);
                DLOG("新连接建立成功！");
            }
            else
            {
                _connection_manager->delConnection(conn);
                LOG_INFO << "连接即将关闭！";
            }
        }

    private:
        muduo::net::EventLoop _loop;
        muduo::net::TcpServer _server;
        ProtobufDispatcher _dispatcher; // 函数分发器
        ProtobufCodecPtr _codec;           // protobuf解释器
        VirtualHost::ptr _host;
        ConsumerManager::ptr _consumers;
        ConnectionManager::ptr _connection_manager;
        Threadpool::ptr _pool;
    };
}
#endif
